package kafka

import (
	"context"
	"fmt"
	"strconv"

	"github.com/Shopify/sarama"
	"github.com/sirupsen/logrus"

	"logproject/model"
)

type KafkaObj struct {
	client  sarama.SyncProducer
	msgChan chan *sarama.ProducerMessage
	cancel  context.CancelFunc
	ctx     context.Context
}

type Registerpc map[string]*sarama.PartitionConsumer

//var client sarama.SyncProducer

func NewKfkaClient() *KafkaObj {
	ctx, cancel := context.WithCancel(context.Background())
	return &KafkaObj{cancel: cancel, ctx: ctx}
}

func Init(address []string, chanSize int64, collectEntryList []model.CollerctEntry) (kafkaobj *KafkaObj, err error, registerpcs Registerpc) {
	//生产者配置
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          //设置等待全部分组成员同步完成
	config.Producer.Partitioner = sarama.NewRandomPartitioner //数据发送到随机分区
	config.Producer.Return.Successes = true

	// 连接kafka
	kafkaobj = NewKfkaClient()
	kafkaobj.client, err = sarama.NewSyncProducer(address, config)
	if err != nil {
		return
	}
	kafkaobj.msgChan = make(chan *sarama.ProducerMessage, chanSize)
	err, registerpcs = kafkaobj.NewPartitionConsumer(collectEntryList, registerpcs)
	if err != nil {
		return
	}
	go kafkaobj.SendMsg()
	return
}

//从MsgChan中读取数据 发送给kafka
func (ka *KafkaObj) SendMsg() {
	for {
		select {
		case msg := <-ka.msgChan:
			_, _, err := ka.client.SendMessage(msg)
			if err != nil {
				logrus.Warning("send msg to kafka err:", err)
				continue
			}
			logrus.Info("send msg to kafka success")
		}
	}
}

//创建分区消费者
func (ka *KafkaObj) NewPartitionConsumer(collectEntryList []model.CollerctEntry, registerPC Registerpc) (error, Registerpc) {
	newRegisterpc := make(Registerpc)
	consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
	if err != nil {
		return err, registerPC
	}

	for _, collectEntry := range collectEntryList {
		partitionList, err := consumer.Partitions(collectEntry.Topic) //根据topic 获取所有分区
		if err != nil {
			return err, registerPC
		}
		for partition := range partitionList { //遍历分区
			//为每个分区 创建一个消费者
			//已有消费者
			if _, ok := registerPC[strconv.Itoa(partition)+collectEntry.Topic]; ok {
				newRegisterpc[strconv.Itoa(partition)+collectEntry.Topic] = registerPC[strconv.Itoa(partition)+collectEntry.Topic]
				continue
			}
			//新消费者
			pc, err := consumer.ConsumePartition(collectEntry.Topic, int32(partition), sarama.OffsetNewest)
			if err != nil {
				return err, registerPC
			}
			//pc.AsyncClose()
			go ka.ConsumerWork(&pc)
			newRegisterpc[strconv.Itoa(partition)+collectEntry.Topic] = &pc

		}
	}
	for key, v := range registerPC {
		found := false
		for newkey := range newRegisterpc {
			if key == newkey {
				found = true
				break
			}
		}
		if !found {
			(*v).AsyncClose()
			registerPC = newRegisterpc
			logrus.Info("stopped consumer")
		}
	}
	//registerPC = newRegisterpc
	return nil, newRegisterpc
}

//消费信息
func (ka *KafkaObj) ConsumerWork(pc *sarama.PartitionConsumer) {
	//var msg = &sarama.ConsumerMessage{}
	//for {
	//	select {
	//	case <-ka.ctx.Done():
	//		return
	//	default:
	//		msg = <-(*pc).Messages()
	//
	//	}
	//}

	for msg := range (*pc).Messages() {
		fmt.Printf("Partition:%d offset:%d key:%s value:%s\n", msg.Partition, msg.Offset, msg.Key, msg.Value)
	}

}

//传入msgChan
func (ka *KafkaObj) ToMsgChan(msg *sarama.ProducerMessage) {
	ka.msgChan <- msg
}

//func ChangeKafkaConsumerTasks(tailfstructForSave *tailfile.Tailf, kafkaobj *KafkaObj, registerpcs registerpc) {
//		newRegisterpc := make(registerpc)
//	if len(tailfstructForSave.EtcdConf) != 0 {
//		for _, filepath := range tailfstructForSave.EtcdConf {
//			//原来已有任务
//			if _, ok := registerpcs[filepath.Topic]; ok {
//				newRegisterpc[filepath.Topic] = registerpcs[filepath.Topic]
//				continue
//			}
//			//新任务
//			tai := newTailf()
//			var err error
//			tai.tailObj, err = tail.TailFile(filepath.Path, config)
//			newRegisterTask[filepath.Topic] = tai
//			registerTasks = newRegisterTask
//			if err != nil {
//				logrus.Errorf("TailFile failed path:%s\n", filepath.Path)
//				continue
//			}
//			go tai.Run(filepath.Topic, kafkaobj)
//		}
//	}
//	for key, v := range registerTasks {
//		found := false
//		for newkey := range newRegisterTask {
//			if key == newkey {
//				found = true
//				break
//			}
//		}
//		if !found {
//			v.cancel()
//			registerTasks = newRegisterTask
//			err := v.tailObj.Stop()
//			if err != nil {
//				logrus.Errorf("stop tailfile failed", err)
//				return
//			}
//			logrus.Info("stopped taifile")
//		}
//	}
//}
